#include "market_data/market_data_storage.h"
#include <filesystem>
#include <fstream>
#include <sstream>
#include <iomanip>
#include <algorithm>
#include <spdlog/spdlog.h>

namespace hft {

namespace fs = std::filesystem;

// CSV文件存储实现
CsvMarketDataStorage::CsvMarketDataStorage(
    const std::string& base_dir,
    int flush_interval)
    : base_dir_(base_dir), flush_interval_(flush_interval) {
    
    // 确保基础目录存在
    ensureDirectoryExists(base_dir_);
    
    // 启动刷新线程
    if (flush_interval_ > 0) {
        flush_thread_ = std::thread(&CsvMarketDataStorage::flushThreadFunc, this);
    }
}

CsvMarketDataStorage::~CsvMarketDataStorage() {
    // 停止刷新线程
    if (flush_thread_.joinable()) {
        {
            std::lock_guard<std::mutex> lock(flush_mutex_);
            running_ = false;
        }
        flush_cv_.notify_one();
        flush_thread_.join();
    }
    
    // 刷新所有缓存
    flushCache();
}

bool CsvMarketDataStorage::store(const MarketDataMessage& data) {
    std::lock_guard<std::mutex> lock(cache_mutex_);
    
    // 获取文件路径
    std::string file_key = data.symbol + "_" + 
                         std::to_string(std::chrono::duration_cast<std::chrono::hours>(
                             data.timestamp.time_since_epoch()).count());
    
    // 检查缓存
    auto it = file_cache_.find(file_key);
    if (it == file_cache_.end()) {
        // 创建新的缓存项
        std::string file_path = getFilePath(data.symbol, data.timestamp);
        bool is_new_file = !fs::exists(file_path);
        
        FileCache cache;
        cache.file.open(file_path, std::ios::app);
        if (!cache.file.is_open()) {
            spdlog::error("Failed to open file for writing: {}", file_path);
            return false;
        }
        
        // 如果是新文件，写入标题行
        if (is_new_file) {
            cache.file << "timestamp,symbol,last_price,volume,turnover,open_interest,"
                      << "bid_price,bid_volume,ask_price,ask_volume\n";
        }
        
        cache.last_flush = std::chrono::system_clock::now();
        it = file_cache_.emplace(file_key, std::move(cache)).first;
    }
    
    // 添加到缓存
    it->second.buffer.push_back(data);
    
    // 如果不使用刷新线程，则立即刷新
    if (flush_interval_ <= 0 && it->second.buffer.size() >= 100) {
        // 写入文件
        for (const auto& msg : it->second.buffer) {
            auto time_t = std::chrono::system_clock::to_time_t(msg.timestamp);
            std::tm tm = *std::localtime(&time_t);
            
            it->second.file << std::put_time(&tm, "%Y-%m-%d %H:%M:%S") << ","
                          << msg.symbol << ","
                          << msg.last_price << ","
                          << msg.volume << ","
                          << msg.turnover << ","
                          << msg.open_interest << ","
                          << msg.bid_price << ","
                          << msg.bid_volume << ","
                          << msg.ask_price << ","
                          << msg.ask_volume << "\n";
        }
        
        it->second.file.flush();
        it->second.buffer.clear();
        it->second.last_flush = std::chrono::system_clock::now();
    }
    
    total_records_++;
    return true;
}

int CsvMarketDataStorage::storeBatch(const std::vector<MarketDataMessage>& data_list) {
    int success_count = 0;
    for (const auto& data : data_list) {
        if (store(data)) {
            success_count++;
        }
    }
    return success_count;
}

std::vector<MarketDataMessage> CsvMarketDataStorage::query(
    const std::string& symbol,
    const std::chrono::system_clock::time_point& start_time,
    const std::chrono::system_clock::time_point& end_time) {
    
    std::vector<MarketDataMessage> results;
    
    // 计算开始和结束的小时
    auto start_hour = std::chrono::duration_cast<std::chrono::hours>(
        start_time.time_since_epoch()).count();
    auto end_hour = std::chrono::duration_cast<std::chrono::hours>(
        end_time.time_since_epoch()).count();
    
    // 遍历每个小时
    for (auto hour = start_hour; hour <= end_hour; ++hour) {
        auto timestamp = std::chrono::system_clock::time_point(std::chrono::hours(hour));
        std::string file_path = getFilePath(symbol, timestamp);
        
        if (!fs::exists(file_path)) {
            continue;
        }
        
        // 读取文件
        std::ifstream file(file_path);
        if (!file.is_open()) {
            spdlog::error("Failed to open file for reading: {}", file_path);
            continue;
        }
        
        // 跳过标题行
        std::string line;
        std::getline(file, line);
        
        // 读取数据行
        while (std::getline(file, line)) {
            std::stringstream ss(line);
            std::string field;
            
            // 解析时间戳
            std::getline(ss, field, ',');
            std::tm tm = {};
            std::stringstream time_ss(field);
            time_ss >> std::get_time(&tm, "%Y-%m-%d %H:%M:%S");
            auto timestamp = std::chrono::system_clock::from_time_t(std::mktime(&tm));
            
            // 检查时间范围
            if (timestamp < start_time || timestamp > end_time) {
                continue;
            }
            
            MarketDataMessage msg;
            msg.timestamp = timestamp;
            
            // 解析其他字段
            std::getline(ss, field, ',');
            msg.symbol = field;
            
            std::getline(ss, field, ',');
            msg.last_price = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.volume = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.turnover = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.open_interest = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.bid_price = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.bid_volume = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.ask_price = std::stod(field);
            
            std::getline(ss, field, ',');
            msg.ask_volume = std::stod(field);
            
            results.push_back(msg);
        }
    }
    
    // 按时间排序
    std::sort(results.begin(), results.end(), 
             [](const MarketDataMessage& a, const MarketDataMessage& b) {
                 return a.timestamp < b.timestamp;
             });
    
    return results;
}

int CsvMarketDataStorage::cleanup(const std::chrono::system_clock::time_point& before_time) {
    int removed_count = 0;
    
    // 计算截止日期
    auto cutoff_day = std::chrono::duration_cast<std::chrono::hours>(
        before_time.time_since_epoch()).count() / 24;
    
    try {
        // 遍历基础目录
        for (const auto& entry : fs::recursive_directory_iterator(base_dir_)) {
            if (!entry.is_regular_file() || entry.path().extension() != ".csv") {
                continue;
            }
            
            // 从文件名中提取日期
            std::string filename = entry.path().filename().string();
            size_t date_pos = filename.find_last_of('_');
            if (date_pos == std::string::npos) {
                continue;
            }
            
            std::string date_str = filename.substr(date_pos + 1, 8); // YYYYMMDD
            if (date_str.length() != 8) {
                continue;
            }
            
            // 解析日期
            std::tm tm = {};
            tm.tm_year = std::stoi(date_str.substr(0, 4)) - 1900;
            tm.tm_mon = std::stoi(date_str.substr(4, 2)) - 1;
            tm.tm_mday = std::stoi(date_str.substr(6, 2));
            
            auto file_time = std::chrono::system_clock::from_time_t(std::mktime(&tm));
            auto file_day = std::chrono::duration_cast<std::chrono::hours>(
                file_time.time_since_epoch()).count() / 24;
            
            // 如果文件日期早于截止日期，则删除
            if (file_day < cutoff_day) {
                // 先检查文件是否在缓存中
                std::lock_guard<std::mutex> lock(cache_mutex_);
                for (auto it = file_cache_.begin(); it != file_cache_.end(); ) {
                    if (it->second.file.is_open() && 
                        entry.path().string().find(it->first) != std::string::npos) {
                        it->second.file.close();
                        it = file_cache_.erase(it);
                    } else {
                        ++it;
                    }
                }
                
                // 删除文件
                fs::remove(entry.path());
                removed_count++;
            }
        }
    } catch (const std::exception& e) {
        spdlog::error("Error during cleanup: {}", e.what());
    }
    
    return removed_count;
}

uint64_t CsvMarketDataStorage::getStorageStats() const {
    return total_records_.load();
}

std::string CsvMarketDataStorage::getFilePath(
    const std::string& symbol,
    const std::chrono::system_clock::time_point& timestamp) const {
    
    // 格式化日期
    auto time_t = std::chrono::system_clock::to_time_t(timestamp);
    std::tm tm = *std::localtime(&time_t);
    
    std::stringstream ss;
    ss << base_dir_ << "/" << symbol << "/";
    ss << std::put_time(&tm, "%Y/%m");
    
    std::string dir_path = ss.str();
    ensureDirectoryExists(dir_path);
    
    ss << "/" << symbol << "_" << std::put_time(&tm, "%Y%m%d") << ".csv";
    return ss.str();
}

bool CsvMarketDataStorage::ensureDirectoryExists(const std::string& path) const {
    try {
        if (!fs::exists(path)) {
            fs::create_directories(path);
        }
        return true;
    } catch (const std::exception& e) {
        spdlog::error("Failed to create directory {}: {}", path, e.what());
        return false;
    }
}

void CsvMarketDataStorage::flushCache() {
    std::lock_guard<std::mutex> lock(cache_mutex_);
    
    for (auto& [key, cache] : file_cache_) {
        if (!cache.buffer.empty() && cache.file.is_open()) {
            for (const auto& msg : cache.buffer) {
                auto time_t = std::chrono::system_clock::to_time_t(msg.timestamp);
                std::tm tm = *std::localtime(&time_t);
                
                cache.file << std::put_time(&tm, "%Y-%m-%d %H:%M:%S") << ","
                          << msg.symbol << ","
                          << msg.last_price << ","
                          << msg.volume << ","
                          << msg.turnover << ","
                          << msg.open_interest << ","
                          << msg.bid_price << ","
                          << msg.bid_volume << ","
                          << msg.ask_price << ","
                          << msg.ask_volume << "\n";
            }
            
            cache.file.flush();
            cache.buffer.clear();
            cache.last_flush = std::chrono::system_clock::now();
        }
    }
}

void CsvMarketDataStorage::flushThreadFunc() {
    while (running_) {
        // 等待刷新间隔或通知
        {
            std::unique_lock<std::mutex> lock(flush_mutex_);
            flush_cv_.wait_for(lock, std::chrono::seconds(flush_interval_),
                              [this] { return !running_; });
        }
        
        if (!running_) {
            break;
        }
        
        // 刷新缓存
        flushCache();
    }
}

// 异步存储实现
AsyncMarketDataStorage::AsyncMarketDataStorage(
    std::shared_ptr<MarketDataStorage> storage,
    size_t queue_size)
    : storage_(storage), max_queue_size_(queue_size) {
    
    // 启动工作线程
    worker_thread_ = std::thread(&AsyncMarketDataStorage::workerThreadFunc, this);
}

AsyncMarketDataStorage::~AsyncMarketDataStorage() {
    // 停止工作线程
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        running_ = false;
    }
    queue_cv_.notify_one();
    
    if (worker_thread_.joinable()) {
        worker_thread_.join();
    }
}

bool AsyncMarketDataStorage::store(const MarketDataMessage& data) {
    std::lock_guard<std::mutex> lock(queue_mutex_);
    
    // 检查队列大小
    if (max_queue_size_ > 0 && data_queue_.size() >= max_queue_size_) {
        dropped_count_++;
        return false;
    }
    
    // 添加到队列
    data_queue_.push(data);
    queue_cv_.notify_one();
    return true;
}

size_t AsyncMarketDataStorage::getQueueSize() const {
    std::lock_guard<std::mutex> lock(queue_mutex_);
    return data_queue_.size();
}

uint64_t AsyncMarketDataStorage::getStorageStats() const {
    return storage_->getStorageStats();
}

bool AsyncMarketDataStorage::waitForCompletion(int timeout_ms) {
    std::unique_lock<std::mutex> lock(queue_mutex_);
    
    if (data_queue_.empty()) {
        return true;
    }
    
    if (timeout_ms <= 0) {
        // 无限等待
        queue_cv_.wait(lock, [this] { return data_queue_.empty() || !running_; });
    } else {
        // 有超时的等待
        queue_cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
                          [this] { return data_queue_.empty() || !running_; });
    }
    
    return data_queue_.empty();
}

void AsyncMarketDataStorage::workerThreadFunc() {
    std::vector<MarketDataMessage> batch;
    batch.reserve(100); // 预分配空间，提高效率
    
    while (running_ || !data_queue_.empty()) {
        batch.clear();
        
        // 获取一批数据
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            
            if (data_queue_.empty() && running_) {
                // 等待新数据
                queue_cv_.wait(lock, [this] { return !data_queue_.empty() || !running_; });
            }
            
            // 如果队列为空且不再运行，则退出
            if (data_queue_.empty() && !running_) {
                break;
            }
            
            // 从队列中取出一批数据
            while (!data_queue_.empty() && batch.size() < 100) {
                batch.push_back(std::move(data_queue_.front()));
                data_queue_.pop();
            }
            
            // 如果队列已空，通知等待的线程
            if (data_queue_.empty()) {
                queue_cv_.notify_all();
            }
        }
        
        // 批量存储数据
        if (!batch.empty()) {
            int stored = storage_->storeBatch(batch);
            processed_count_ += stored;
            
            if (stored < static_cast<int>(batch.size())) {
                dropped_count_ += (batch.size() - stored);
            }
        }
    }
}

} // namespace hft